Skip to content

feat(message-bus)!: rewrite around class-based events and per-broker queues - #878

Open
johngeorgewright wants to merge 19 commits into
masterfrom
dmg-bus
Open

johngeorgewright wants to merge 19 commits into
masterfrom
dmg-bus

Conversation

@johngeorgewright

@johngeorgewright johngeorgewright commented Sep 17, 2026

Copy link
Copy Markdown
Collaborator

Replace the old MessageBus API (string event names, tuple args, generators and
RPC-style invokables) with a class-based bus modelled on Enterprise Integration
Patterns. Messages are plain class instances routed by constructor identity, so
EventDescriptions are no longer needed. The invokable and generator APIs collapse
into a single streaming register/invoke pair, and the RPC-style request/response
model is removed.

Each bus member is a Participant with its own queue, so members can be paused
and resumed independently. A participant has two faces onto the same kernel: a
MessageGateway (outbound — what the application holds, to subscribe, emit and
invoke) and a MessageDispatcher (inbound — what the bus drives to deliver
messages to the registered handlers). The handler roles are named after their EIP
counterparts: Subscribers (on/once/until), Responders (register) that
stream values back for a CommandMessage, and Interceptors (intercept) that form
a Message Translator / Pipes-and-Filters chain able to transform or CANCEL a
message before subscribers see it. Filters are a Message Filter (whole-message
pass/discard).

Changes since this description was first written

The branch has moved on considerably since the PR was opened:

  • Enterprise Integration Patterns rename: Bus → MessageBus, broker →
    participant (facade: MessageGateway), Event → Message, Invocation →
    CommandMessage, and handlers are now Subscriber / Responder / Interceptor.
  • Externally abortable participants: bus.gateway(name, signal) ties a
    participant's lifetime to an AbortSignal; aborting tears down its handlers and
    frees its name for reuse.
  • Serializable payloads: emit/invoke arguments are constrained at compile
    time to JSON-safe data. Because a message can sit in a paused queue and be
    delivered later, a payload holding a live reference (e.g. an HTMLElement)
    would no longer be a faithful snapshot; the constraint guarantees a subscriber
    receives what was emitted, whenever it is emitted. It is a type-level rule
    only — nothing is cloned at runtime.

Bug fixes found while reviewing the rewrite

Each is covered by tests:

  • filters now AND their keys instead of OR
  • each emit() call resolves to its own result rather than a shared promise
  • invoke completion is the settling of each responder, not a finish() count, so a
    filtered-out or negligent responder can no longer hang or truncate the stream
  • invoke gains an error channel: fail-loud by default, isolate with { onError }
  • interceptor routing is cleaned up when a participant aborts
  • queueMethod no longer resets a running queue

BREAKING CHANGE: the entire public API of @plugola/message-bus has changed.

  • MessageBus replaces the old bus; bus.gateway(name) returns a MessageGateway,
    optionally scoped to an AbortSignal. EventDescriptions are removed.
  • Messages are classes implementing Message (or extending CommandMessage) and
    are subscribed to by class, not string name.
  • Invokables and generators are unified into a streaming API: register a responder
    with gateway.register(Command, (command, { send, signal }) => ...) and call
    gateway.invoke(new Command()).collect() / .iterate(). There is no finish() — a
    responder completes when it returns or its promise settles. The RPC-style
    request/response model is removed.
  • invoke errors are fail-loud by default; pass { onError } to invoke() to isolate
    responders and keep the stream running.
  • Message and command payloads must be Serializable (JSON-safe), enforced at the
    type level on emit/invoke.
  • Lifecycle methods are renamed: start/stop are now resume/pause on both
    MessageBus and MessageGateway, with abort for permanent teardown.
  • CancelEvent is renamed to CANCEL.

🤖 Generated with Claude Code

johngeorgewright and others added 19 commits September 4, 2026 17:24
We only need the one.
…queues

Replace the MessageBus API (string event names, tuple args, generators and
RPC-style invokables) with a class-based bus. Events are plain class instances
routed by constructor identity, so EventDescriptions are no longer needed. The
invokable and generator APIs collapse into a single streaming register/invoke
pair, and the RPC-style request/response model is removed. Each broker owns its
own queue, so brokers can be paused and resumed independently.

Also fixes flaws found while reviewing the rewrite, each covered by tests:
- filters now AND their keys instead of OR
- each emit() call resolves to its own result rather than a shared promise
- invoke completion is the settling of each handler, not a finish() count, so a
  filtered-out or negligent handler can no longer hang or truncate the stream
- invoke gains an error channel: fail-loud by default, isolate with { onError }
- interceptor routing is cleaned up when a broker aborts
- queueMethod no longer resets a running queue

BREAKING CHANGE: the entire public API of @plugola/message-bus has changed.

- `MessageBus` is replaced by `Bus`; `bus.broker(name)` returns a `PluginBroker`.
- Events are classes implementing `Event` (or extending `Invocation<T>`) and are
  subscribed to by class, not string name. `EventDescription`s are removed.
- Invokables and generators are unified into a streaming API: register a handler
  with `broker.register(Invocation, (event, { send, signal }) => ...)` and call
  `broker.invoke(new Invocation()).collect()` / `.iterate()`. There is no
  `finish()` — a handler completes when it returns or its promise settles. The
  RPC-style request/response model is removed.
- Invocation errors are fail-loud by default; pass `{ onError }` to `invoke()`
  to isolate handlers and keep the stream running.
- Lifecycle methods are renamed: `start`/`stop` are now `resume`/`pause` on both
  `Bus` and `PluginBroker`.
- `CancelEvent` is renamed to `CANCEL`.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Optionally pass an abort signal in to broker construction.
Realign the package's vocabulary with the EIP glossary:

- Bus            -> MessageBus
- PluginBroker   -> MessageGateway   (a participant's endpoint facade)
- Broker         -> MessageDispatcher (drives local performers)
- Event          -> Message
- Invocation     -> CommandMessage
- EventListener        -> Subscriber
- InvocationListener   -> Responder
- InterceptionListener -> Interceptor
- InvocationType -> ResponseType; $invocationType -> $responseType
- bus.broker(name) -> bus.gateway(name)

The dispatcher's inbound methods now say what they do rather than
mirroring the gateway's outbound names: emit -> dispatch, invoke ->
dispatchCommand, intercept -> runInterceptors.

Group modules by concept: Message/ (Message, CommandMessage), Roles/
(Subscriber, Responder, Interceptor), Gateway/ (MessageGateway,
MessageDispatcher, Handler). Document the EIP pattern each concept
maps to (Message Filter, Selective Consumer, Pipes-and-Filters).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Constrain emit/invoke arguments to a JSON-safe Serializable type so a
message that holds a live reference (e.g. an HTMLElement) is a compile
error. Messages can sit in a paused queue and be delivered later, so a
payload must be a faithful snapshot rather than a reference that may be
detached or mutated by then.

The check is type-level only; $-prefixed metadata keys ($name,
$responseType) are exempt since the class wrapper is never serialized.

Add tsc-gated type tests (test:types) covering the mapping and the
emit/invoke call sites.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The gateway and dispatcher were coupled parent/child: the gateway reached
through `dispatcher.bus` to publish and into the dispatcher's registries to
wire up performers, while the dispatcher held `bus`/`name` purely for the
gateway's benefit.

Introduce a `Participant` that owns the shared kernel — name, queue, abort
lifecycle, and a `PerformerRegistry` — with two thin faces onto it:
`MessageGateway` (outbound) writes the registry and publishes on the bus;
`MessageDispatcher` (inbound) reads the registry and is gated on the queue.
Neither half owns the other, and the ownership graph now matches the runtime
flow (gateway.emit -> bus -> dispatcher). The dispatcher no longer holds the
bus at all.

Also rename `Handler` -> `SelectivePerformer` (`handle` -> `perform`) and the
`Gateway/` directory -> `Participant/`, aligning the whole folder with EIP
endpoint vocabulary (Messaging Gateway, Message Dispatcher, Selective
Consumer).

Note: on abort the registry now clears interceptors too, not just subscribers
and responders. This is unobservable — the bus drops the participant from all
routing and from its map at the same time — but is called out for review.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant